// // consumer.cpp
// #include <iostream>
// #include <signal.h>
// #include <cstdlib>
// #include <librdkafka/rdkafkacpp.h>

// class ExampleConsumeCallback : public RdKafka::ConsumeCb {
// public:
//     void consume_cb(RdKafka::Message &msg, void *opaque) override {
//         if (msg.err()) {
//             std::cerr << "Consume error: " << msg.errstr() << std::endl;
//             return;
//         }

//         std::cout << "Received message: " << std::string(static_cast<const char *>(msg.payload()), msg.len())
//                   << " from topic " << msg.topic_name()
//                   << " [" << msg.partition() << "] at offset " << msg.offset()
//                   << std::endl;
//     }
// };

// int main(int argc, char **argv) {
//     std::string brokers = "localhost:9092";
//     std::string topic_str = "test-topic";
//     std::string group_id = "cpp-consumer-group";

//     // 创建全局配置对象
//     RdKafka::Conf *conf = RdKafka::Conf::create(RdKafka::Conf::CONF_GLOBAL);
//     std::string errstr;

//     // 设置 Broker 地址
//     if (conf->set("bootstrap.servers", brokers, errstr) != RdKafka::Conf::CONF_OK) {
//         std::cerr << errstr << std::endl;
//         exit(1);
//     }

//      // 设置消费者组 ID
//     if (conf->set("group.id", group_id, errstr) != RdKafka::Conf::CONF_OK) {
//         std::cerr << errstr << std::endl;
//         exit(1);
//     }

//     if (conf->set("auto.commit.interval.ms", "1000", errstr) != RdKafka::Conf::CONF_OK) {
//         std::cerr << errstr << std::endl;
//         exit(1);
//     }

//     ExampleConsumeCallback consume_cb;
//     conf->set("consume_cb", &consume_cb, errstr);

//     // 2. 创建消费者实例
//     RdKafka::Consumer *consumer = RdKafka::Consumer::create(conf, errstr);
//     if (!consumer) {
//         std::cerr << "Failed to create consumer: " << errstr << std::endl;
//         exit(1);
//     }
//     // 3. 创建主题
//     RdKafka::Topic *topic = RdKafka::Topic::create(consumer, topic_str, NULL, errstr);
//     if (!topic) {
//         std::cerr << "Failed to create topic object: " << errstr << std::endl;
//         exit(1);
//     }


//     // RdKafka::ErrorCode resp = consumer->start(topic, RdKafka::Topic::PARTITION_UA, RdKafka::Topic::OFFSET_BEGINNING);
//     // if (resp != RdKafka::ERR_NO_ERROR) {
//     //     std::cerr << "Failed to start consumer: " << RdKafka::err2str(resp) << std::endl;
//     //     exit(1);
//     // }

//      // 4. 拉取消息并消费
//     std::cout << "Consumer started. Waiting for messages..." << std::endl;
//     do {
//         consumer->poll(1000); // 超时时间 1000ms
//     } while (true);

//     consumer->stop(topic, RdKafka::Topic::PARTITION_UA);
//     delete topic;
//     delete consumer;
//     delete conf;

//     return 0;
// }

#include <librdkafka/rdkafkacpp.h>
#include <iostream>
#include <csignal>
#include <memory>

bool running = true;

void signal_handler(int) {
  running = false;
}

class ExampleEventCb : public RdKafka::EventCb {
  void event_cb(RdKafka::Event &event) override {
    if (event.type() == RdKafka::Event::EVENT_ERROR) {
      std::cerr << "Kafka Error: " << event.str() << std::endl;
    }
  }
};

int main() {
  signal(SIGINT, signal_handler);

  std::string brokers = "localhost:9092";
  std::string topic = "test_topic";
  std::string group_id = "cpp_consumer_group";

  std::string errstr;
  ExampleEventCb event_cb;

  auto conf = RdKafka::Conf::create(RdKafka::Conf::CONF_GLOBAL);
  conf->set("bootstrap.servers", brokers, errstr);
  conf->set("group.id", group_id, errstr);
  conf->set("auto.offset.reset", "earliest", errstr);
  conf->set("event_cb", &event_cb, errstr);

  auto consumer = RdKafka::KafkaConsumer::create(conf, errstr);
  if (!consumer) {
    std::cerr << "Failed to create consumer: " << errstr << std::endl;
    return 1;
  }

  consumer->subscribe({topic});
  std::cout << "Consuming messages from topic " << topic << std::endl;

  while (running) {
    auto msg = consumer->consume(1000);
    if (msg->err() == RdKafka::ERR_NO_ERROR) {
      std::string message(reinterpret_cast<const char*>(msg->payload()), msg->len());
		std::cout << "Received message: " << message << std::endl;
    }
    delete msg;
  }

  consumer->close();
  delete consumer;
  return 0;
}


